package com.chengzhi.nettycommoms.disruptor;

import com.chengzhi.nettycommoms.dto.NettyMessage;
import com.chengzhi.nettycommoms.dto.NettyMessageWapper;
import com.lmax.disruptor.RingBuffer;
import io.netty.channel.ChannelHandlerContext;

public class MessageProducer {

    private RingBuffer<NettyMessageWapper> ringBuffer;

    private String producerId;

    public MessageProducer(String producerId, RingBuffer<NettyMessageWapper> ringBuffer) {
        this.ringBuffer = ringBuffer;
        this.producerId = producerId;
    }

    public void send(NettyMessage nettyMessage, ChannelHandlerContext ctx) {
        long sequence = this.ringBuffer.next();
        try {
            NettyMessageWapper messageWapper = ringBuffer.get(sequence);
            messageWapper.setCtx(ctx);
            messageWapper.setNettyMessage(nettyMessage);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}
